import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.graphx.{Edge, EdgeContext, Graph, VertexRDD}

/**
  * Created by danke on 2020/4/18.
  */
object Graphx_Aggregation {
  def sendMsg(ec: EdgeContext[Int, String, Int]): Unit = {
    ec.sendToDst(ec.srcAttr + 1)
  }

  def mergeMsg(a: Int, b: Int): Int = {
    math.max(a, b)
  }

  def sumEdgeCount(g: Graph[Int, String], num: Int): Graph[Int, String] = {
    val verts: VertexRDD[Int] = g.aggregateMessages[Int](sendMsg, mergeMsg)
    val g2 = Graph(verts, g.edges)

    println("-----------------------------------------")
    // g2.vertices.map(x=>x._2).collect.foreach(println(_))
    println(g2.vertices.map(x=>x._2).max())
    println("-----------------------------------------")
    //定义len进行判断
    val len: Int = g2.vertices.map(x => x._2).max()
    if (len > num)
      sumEdgeCount(g2, num + 1)
    else
      g
  }

  def main(args: Array[String]): Unit = {
    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    // 构建图
    val myVertices =
      sc.parallelize(Array((1L, "张三"), (2L, "李四"), (3L, "王五"), (4L, "钱六"),
        (5L, "领导")))
    val myEdges =
      sc.parallelize(Array( Edge(1L,2L,"朋友"),
        Edge(2L,3L,"朋友") , Edge(3L,4L,"朋友"),
        Edge(4L,5L,"上下级"),Edge(3L,5L,"上下级")
      ))
    val myGraph = Graph(myVertices, myEdges)

    //将顶点的属性集初始化为0
    val initGraph = myGraph.mapVertices((_,_)=>0)

    sumEdgeCount(initGraph, 0).vertices.collect.foreach(println(_))
  }
}
